package com.ksyun.campus.client.util;

import com.alibaba.fastjson2.JSON;
import com.ksyun.campus.client.common.constant.ServerConfiguration;
import com.ksyun.campus.client.common.context.ServerContext;
import com.ksyun.campus.client.domain.server.DataServer;
import com.ksyun.campus.client.domain.server.MetaServer;
import lombok.extern.slf4j.Slf4j;
import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.PathChildrenCache;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;

import javax.annotation.PostConstruct;
import java.util.List;
import java.util.stream.Collectors;

@Slf4j
public class ZkUtil {
    public CuratorFramework client;
    public Boolean isConnect = false;
    public Boolean isClose = false;
    public String zkConnectionString = ServerConfiguration.ZKPATH;
    public String zkMetaServerPath = ServerConfiguration.ZKMETASERVERPATH;
    public String zkDataServerPath = ServerConfiguration.ZKDATASERVERPATH;

    public ZkUtil() {
        try {
            postCons();
        } catch (Exception e) {
            log.error("Error initializing ZkUtil", e);
            throw new RuntimeException("Failed to initialize ZkUtil", e);
        }
    }

    //初始化，与zk建立连接，注册监听路径，当配置有变化随时更新
    @PostConstruct
    public void postCons() throws Exception {
        if (!isConnect) {
            isConnect = true;
            //todo 参数的设置怎么更改 初始化的问题？？？？？
            RetryPolicy retryPolicy = new ExponentialBackoffRetry(3000, 10);
            client = CuratorFrameworkFactory.builder()
                    .connectString(zkConnectionString)
                    .sessionTimeoutMs(60 * 1000)
                    .connectionTimeoutMs(15 * 1000)
                    .retryPolicy(retryPolicy).build();
            client.start();
            System.out.println("zkTtil:  建立连接成功:::::::::::::::");

            //连接zookeeper时,获取到主从MetaServer的信息
            List<String> serverStringList = client.getChildren().forPath(zkMetaServerPath);
            List<MetaServer> metaServerList = serverStringList.stream().map(each -> {
                try {
                    String childrenPath = zkMetaServerPath + "/" + each;
                    byte[] MetaServerBytes = client.getData().forPath(childrenPath);
                    return JSON.parseObject(MetaServerBytes, MetaServer.class);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }).collect(Collectors.toList());
            System.out.println("metaServer列表: " + metaServerList);
            ServerContext.updateMetaServerList(metaServerList);
//            System.out.println("-----------------------------");
            registerWatcher(zkMetaServerPath);


            //连接zookeeper时,获取到主从dataServer的信息
            List<String> dataServerStringList = client.getChildren().forPath(zkDataServerPath);
            List<DataServer> dataServerList = dataServerStringList.stream().map(each -> {
                try {
                    String childrenPath = zkDataServerPath + "/" + each;
                    byte[] DataServerBytes = client.getData().forPath(childrenPath);
                    return JSON.parseObject(DataServerBytes, DataServer.class);
                } catch (Exception e) {
                    throw new RuntimeException(e);
                }
            }).collect(Collectors.toList());
//            System.out.println("dataServer列表: " + dataServerList);
            ServerContext.updateDataServerList(dataServerList);
//            System.out.println("-----------------------------");
            registerWatcher(zkDataServerPath);
        }

    }


    /**
     * 注册对metaServer的监听器 (主从切换会触发)
     * @param path
     * @throws Exception
     */
    private void registerWatcher(String path) throws Exception {
        PathChildrenCache cache = new PathChildrenCache(client, path, true);
        PathChildrenCacheListener listener = new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
//                System.out.println("监听到MetaServer发生事件变化");
                List<String> serverStringList = client.getChildren().forPath(zkMetaServerPath);
                List<MetaServer> metaServerList = serverStringList.stream().map(each -> {
                    try {
                        String childrenPath = zkMetaServerPath + "/" + each;
                        byte[] MetaServerBytes = client.getData().forPath(childrenPath);
//                        System.out.println("从zookeeper中获取MetaServer的信息:" + new String(MetaServerBytes));
                        return JSON.parseObject(MetaServerBytes, MetaServer.class);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }).collect(Collectors.toList());
                ServerContext.updateMetaServerList(metaServerList);
//                System.out.println("更新ThreadLocal中存的metaServer列表: " + ServerContext.getMetaServerList());
//                System.out.println("-----------------------------");

                List<String> dataServerStringList = client.getChildren().forPath(zkDataServerPath);
                List<DataServer> dataServerList = dataServerStringList.stream().map(each -> {
                    try {
                        String childrenPath = zkDataServerPath + "/" + each;
                        byte[] DataServerBytes = client.getData().forPath(childrenPath);
//                        System.out.println("从zookeeper中获取MetaServer的信息:" + new String(DataServerBytes));
                        return JSON.parseObject(DataServerBytes, DataServer.class);
                    } catch (Exception e) {
                        throw new RuntimeException(e);
                    }
                }).collect(Collectors.toList());
//                System.out.println("dataServer列表: " + dataServerList);
                ServerContext.updateDataServerList(dataServerList);
//                System.out.println("-----------------------------");
                registerWatcher(zkDataServerPath);

            }
        };
        cache.getListenable().addListener(listener);
        cache.start();
    }

    public void close() {
        if (!isClose) {
            isClose = true;
            System.out.println("关闭与zookeeper的连接");
            client.close();
        }
    }

}
